#include "pml_hash.h"
#include <sys/mman.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <stdio.h>
#include <errno.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#include <libpmem.h>

#include <iostream>
using namespace std;
/**
 * PMLHash::PMLHash 
 * 
 * @param  {char*} file_path : the file path of data file
 * if the data file exist, open it and recover the hash
 * if the data file does not exist, create it and initial the hash
 */


PMLHash::PMLHash(const char* file_path) {
	int fd;
	/* create a pmem file */
	if ((fd = open(file_path, O_CREAT|O_RDWR, 0666)) < 0) {
		perror("open");
		exit(1);
	}
	/* allocate the pmem */
	if ((errno = posix_fallocate(fd, 0, FILE_SIZE)) != 0) {
		perror("posix_fallocate");
		exit(1);
	}
	/* memory map it */
	if ((start_addr = mmap(NULL, FILE_SIZE, PROT_READ|PROT_WRITE, MAP_SHARED, fd, 0)) == NULL) {
		perror("pmem_map");
		exit(1);
	}
	this->overflow_addr = start_addr + FILE_SIZE/2;
	this->meta = (metadata*)start_addr;
	//the size of meta depend on os,so we decide do this
	this->table_arr = (pm_table*)(this->meta+1);
	// 初始化元数据
	if (meta->size == 0)
	{
		memset(start_addr,0,FILE_SIZE);
		meta->size = HASH_SIZE;
		meta->level = 2;
		meta->next = 0;
		meta->overflow_num = 0;
		msync(start_addr, FILE_SIZE,MS_SYNC);
	}
}
/**
 * PMLHash::~PMLHash 
 * 
 * unmap and close the data file
 */
PMLHash::~PMLHash() {
    munmap(start_addr, FILE_SIZE);
}
/**
 * PMLHash 
 * 
 * split the hash table indexed by the meta->next
 * update the metadata
 */
void PMLHash::split() {
    // fill the split table
    //分裂桶里面的数据全部拿出来重新hash
    //找出所有需要丢进新桶的元素，从旧桶删除
    entry tmp;
    vector<entry> vec_entry;
    pm_table* pt = table_arr + meta->next;
    for (int i = 0; i < pt->fill_num; ++i){
    	tmp = pt->kv_arr[i];
    	uint64_t index = hashFunc(tmp.key, 2 * meta->level);
    	if (index != meta->next){
    		vec_entry.push_back(tmp);
    		remove(tmp.key);
    	}
    }
    while(pt->next_offset != 0){
    	pt = (pm_table*)(start_addr + pt->next_offset);
    	for (int i = 0; i < pt->fill_num; ++i){
	    	tmp = pt->kv_arr[i];
	    	uint64_t index = hashFunc(tmp.key, 2 * meta->level);
	    	if (index != meta->next){
	    		vec_entry.push_back(tmp);
	    		remove(tmp.key);
	    	}
    	}
    }
    // update the next of metadata
	meta->next ++;
	// fill the new table
    //元素插入新桶
    for (int i = 0; i < vec_entry.size(); ++i){
    	insert(vec_entry[i].key,vec_entry[i].value);
    }
    if (meta->next % TABLE_SIZE == 0){
    	meta->next = 0;
    	meta->level *= 2;
    }
	msync(start_addr, FILE_SIZE,MS_SYNC);
}
/**
 * PMLHash 
 * 
 * @param  {uint64_t} key     : key
 * @param  {size_t} hash_size : the N in hash func: idx = hash % N
 * @return {uint64_t}         : index of hash table array
 * 
 * need to hash the key with proper hash function first
 * then calculate the index by N module
 */
uint64_t PMLHash::hashFunc(const uint64_t &key, const size_t &hash_size) {
	return key % hash_size;
}

/**
 * PMLHash 
 * 
 * @param  {uint64_t} offset : the file address offset of the overflow hash table
 *                             to the start of the whole file
 * @return {pm_table*}       : the virtual address of new overflow hash table
 */
pm_table* PMLHash::newOverflowTable(uint64_t &offset) {
	offset = FILE_SIZE/2 + this->meta->overflow_num * sizeof(pm_table);
	this->meta->overflow_num ++;
	msync(start_addr, FILE_SIZE,MS_SYNC);
	return (pm_table*)(start_addr + offset);
}

/**
 * PMLHash 
 * 
 * @param  {uint64_t} key   : inserted key
 * @param  {uint64_t} value : inserted value
 * @return {int}            : success: 0. fail: -1
 * 
 * insert the new kv pair in the hash
 * 
 * always insert the entry in the first empty slot
 * 
 * if the hash table is full then split is triggered
 */
int PMLHash::insert(const uint64_t &key, const uint64_t &value) {
	entry new_entry = {key,value};
	uint64_t index = hashFunc(key,meta->level);
	if (index < meta->next){
		index = hashFunc(key,2*meta->level);
	}
	pm_table* pt = table_arr + index;
	//检查是否以及存在，以及存在就不插入
	while (pt->fill_num == TABLE_SIZE ){
		for (int i = 0; i < TABLE_SIZE; ++i){
			if (pt->kv_arr[i].key == key){
				return -1;
			}
		}
		pt = (pm_table*)(start_addr + pt->next_offset);
	}
	for (int i = 0; i < pt->fill_num; ++i){
		if (pt->kv_arr[i].key == key){
			return -1;
		}
	}
	//直接插入，并fill_num自增
	pt->kv_arr[pt->fill_num] = new_entry;
	pt->fill_num ++;
	msync(start_addr, FILE_SIZE,MS_SYNC);
	//如果桶满了就分裂，并且增加溢出桶
	if (pt->fill_num == TABLE_SIZE)
	{
		split();
		newOverflowTable(pt->next_offset);
		pt = (pm_table*)pt->next_offset;
	}
	return 0;
}

/**
 * PMLHash 
 * 
 * @param  {uint64_t} key   : the searched key
 * @param  {uint64_t} value : return value if found
 * @return {int}            : 0 found, -1 not found
 * 
 * search the target entry and return the value
 */
int PMLHash::search(const uint64_t &key, uint64_t &value) {
	uint64_t index = hashFunc(key,meta->level);
	if (index < meta->next){
		index = hashFunc(key,2*meta->level);
	}
	pm_table* pt = table_arr + index;
	while (pt->next_offset != 0 ){
		for (int i = 0; i < TABLE_SIZE; ++i){
			if (pt->kv_arr[i].key == key){
				value = pt->kv_arr[i].value;
				return 0;
			}
		}
		pt = (pm_table*)(start_addr + pt->next_offset);
	}
	for (int i = 0; i < TABLE_SIZE; ++i){
		if (pt->kv_arr[i].key == key){
			value = pt->kv_arr[i].value;
				return 0;
		}
	}
	return -1;
}

/**
 * PMLHash 
 * 
 * @param  {uint64_t} key : target key
 * @return {int}          : success: 0. fail: -1
 * 
 * remove the target entry, move entries after forward
 * if the overflow table is empty, remove it from hash
 */
int PMLHash::remove(const uint64_t &key) {
	uint64_t index = hashFunc(key,meta->level);
	if (index < meta->next){
		index = hashFunc(key,2*meta->level);
	}
	pm_table* pt = table_arr + index;
	for (int i = 0; i < TABLE_SIZE; ++i){
		if (pt->kv_arr[i].key == key){
			//do remove
			for (int j = i; j < pt->fill_num; ++j){
				//会取到区域外内存，不过无所谓反正不用
				pt->kv_arr[j] = pt->kv_arr[j+1];
			}
			pt->fill_num --;
			msync(start_addr, FILE_SIZE,MS_SYNC);
			//第一个桶不是溢出桶不需要移除
			return 0;
		}
	}
	pm_table* front_pt = pt;

	while (pt->next_offset != 0 ){
		pt = (pm_table*)(start_addr + pt->next_offset);
		for (int i = 0; i < TABLE_SIZE; ++i){
			if (pt->kv_arr[i].key == key){
				//do remove
				for (int j = i; j < pt->fill_num; ++j){
					//会取到区域外内存，不过无所谓反正不用
					pt->kv_arr[j] = pt->kv_arr[j+1];
				}
				pt->fill_num --;
				msync(start_addr, FILE_SIZE,MS_SYNC);
				//如果空了就移除这个桶，从hash桶链表
				if (pt->fill_num == 0)
				{
					front_pt->next_offset = pt->next_offset;
				}
				return 0;
			}
		}
		front_pt = pt;
	}
	return -1;
}

/**
 * PMLHash 
 * 
 * @param  {uint64_t} key   : target key
 * @param  {uint64_t} value : new value
 * @return {int}            : success: 0. fail: -1
 * 
 * update an existing entry
 */
int PMLHash::update(const uint64_t &key, const uint64_t &value) {
	//不存在直接返回失败
	uint64_t index = hashFunc(key,meta->level);
	if (index < meta->next){
		index = hashFunc(key,2*meta->level);
	}
	pm_table* pt = table_arr + index;
	while (pt->next_offset != 0 ){
		for (int i = 0; i < TABLE_SIZE; ++i){
			if (pt->kv_arr[i].key == key){
				pt->kv_arr[i].value = value;
				return 0;
			}
		}
		pt = (pm_table*)(start_addr + pt->next_offset);
	}
	for (int i = 0; i < TABLE_SIZE; ++i){
		if (pt->kv_arr[i].key == key){
				pt->kv_arr[i].value = value;
				return 0;
		}
	}
	return -1;
}

